package com.leilei;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.CURRENT_ROW;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/21 13:21
 */
public class Flink_Table_3_demo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        DataStreamSource<GirlFriend> scFriendList = env.fromCollection(Arrays.asList(new GirlFriend("小丽", "A", 22),
                new GirlFriend("小丽于", "C", 24),
                new GirlFriend("小胡", "D", 23),
                new GirlFriend("小丽于", "D", 17),
                new GirlFriend("小胡2", "D", 25)));
        Table scTable = tableEnv.fromDataStream(scFriendList);
        Table table = scTable.select($("name"), $("cup"), $("age"));
        DataStreamSource<Friend> friendList = env.fromCollection(Arrays.asList(new Friend("小胡","胖虎"),
                new Friend("小丽于","迪迦")));
        Table friendTable = tableEnv.fromDataStream(friendList);
        Table joinTable = table.rightOuterJoin(friendTable, $("name").isEqual($("friendName")));
        DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tableEnv.toRetractStream(joinTable, Row.class);
        tuple2DataStream.print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Friend {
        private String friendName;
        private String nickName;
    }

}
